package ex.datastream.functions.window;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

public class ApplyWindowFunction  implements WindowFunction<Tuple2<String, Long>, String, String, TimeWindow> {
        @Override
        public void apply(String s, TimeWindow timeWindow, Iterable<Tuple2<String, Long>> iterable, Collector<String> collector) throws Exception {
            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            long count = 0;
            for (Tuple2<String, Long> in: iterable) {
                count++;
            }
            String str=String.format("start:%s;end:%s;key:%s;count:%s",
                    dateFormat.format(timeWindow.getStart()),
                    dateFormat.format(timeWindow.getEnd()),s,count);
            collector.collect(str);
        }
}
